package org.yonggan.dmp.report

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 按照低于统计
  *
  * 省市/城市  统计维度
  * 原始请求
  * 有效请求
  * 广告请求
  * 参与竞价数
  * 竞价成功数
  * 竞价成功率
  * 展示量
  * 点击量
  * 点击率
  * 广告成本
  * 广告 消费
  */
object RptADRegionalRDD {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(" 地域分布统计 ")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 创建SqlContext
    val sqlContext = new SQLContext(sc)

    // 先读取文件
    val baseDf = sqlContext.read.parquet("D:\\project\\5")

    val baseRDD: RDD[((String, String), List[Double])] = baseDf.rdd.map(row => {
      // 提取数据
      // 计算维度  省  市
      val pro = row.getAs[String]("provincename")
      val city = row.getAs[String]("cityname")

      //计算出原始请求 requestmode  数据请求方式（1:请求、2:展示、3:点击
      val requestmode = row.getAs[Int]("requestmode")

      //     有效请求数  processnode Int, 流程节点（1：请求量 kpi 2：有效请求 3：广告请求
      val processnode = row.getAs[Int]("processnode")

      //   iseffective: Int, 有效标识（有效指可以正常计费的）(0：无效 1：有效
      val iseffective = row.getAs[Int]("iseffective")

      //   isbilling: Int, 是否收费（0：未收费 1：已收费）
      val isbilling = row.getAs[Int]("isbilling")

      //  isbid: Int, 是否 rtb
      val isbid = row.getAs[Int]("isbid")

      //  iswin: Int, 是否竞价成功
      val iswin = row.getAs[Int]("iswin")

      //   adorderid: Int, 广告 id
      val adorderid = row.getAs[Int]("adorderid")

      // 原始请求数
      val origReq = if (requestmode == 1 && processnode >= 1) 1 else 0
      //  有效请求
      val validReq = if (requestmode == 1 && processnode >= 2) 1 else 0
      // 广告请求数
      val adReq = if (requestmode == 1 && processnode == 3) 1 else 0
      //参与竞价数
      val biddingCnt = if (iseffective == isbilling == isbid == 1 && adorderid != 0) 1 else 0

      val flag = iseffective == isbilling == iswin == 1
      // 竞价成功数
      val succBiddingCnt = if (flag) 1 else 0
      //展示量
      val showCnt = if (requestmode == 2 && iseffective == 1) 1 else 0
      //点击数
      val clickCnt = if (requestmode == 3 && iseffective == 1) 1 else 0
      // DSP广告消费
      val dspConsumpt = if (flag) row.getAs[Double]("winprice") / 1000 else 0d
      //DSP广告成本
      val dspCost = if (flag) row.getAs[Double]("adpayment") / 1000 else 0d

      ((pro, city), List[Double](origReq, validReq, adReq, biddingCnt, succBiddingCnt, showCnt, clickCnt, dspConsumpt, dspCost))
    })

//    baseRDD.foreach(println)

    // 分组统计数据
    val rest: RDD[((String, String), List[Double])] = baseRDD.reduceByKey((lst1, lst2) => {
      val lst: List[(Double, Double)] = lst1.zip(lst2)
      val restult: List[Double] = lst.map(tp => {
        tp._1 + tp._2
      })
      restult
    })

    rest.foreach(println)

  }

}
